package com.webull.information.center.carwler.job.splan;

import java.util.Date;
import java.util.List;
import java.util.Optional;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.support.CronSequenceGenerator;
import org.springframework.stereotype.Component;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.webull.information.center.carwler.common.model.SpeclNewsPlan;
import com.webull.information.center.carwler.common.util.CarwlerJobStateUtil;
import com.webull.information.center.carwler.common.util.eljob.ElasticJobUtil;
import com.webull.information.center.carwler.dao.SpeclNewsPlanDao;
import com.webull.information.center.carwler.job.splan.frame.SplanHandlerMapping;
import com.webull.information.center.carwler.job.splan.frame.SplanTask;

/**
 * 
 * <p>
 * 新闻采集的 特殊调度计划<br>
 * 用于调整指定标的物的采集时间周期，达到改变优先级的目标
 * </p>
 * 
 * @author smj
 * @Date 2017年7月21日上午9:54:22
 * @version 1.0
 *
 */
@Component
public class NewsSpecialPalnJob implements SimpleJob {
	private @Autowired SpeclNewsPlanDao newsSpeclPlanDao;
	private @Autowired SplanHandlerMapping splanHandlerMapping;
	private @Autowired @Qualifier("executor_splan") ThreadPoolTaskExecutor taskExecutor;
	private @Autowired CarwlerJobStateUtil carwlerJobStateUtil;
	protected final Logger logger = LogManager.getLogger(getClass());

	@Override
	public void execute(ShardingContext context) {
		int sharding = Optional.ofNullable(context).map(v -> v.getShardingItem()).orElse(-1);
		int totalCount = Optional.ofNullable(context).map(v -> v.getShardingTotalCount()).orElse(-1);
		logger.info("NewsSpecialPalnJob start:" + ArrayUtils.toString(sharding) + ",totalCount:" + totalCount);
		processTaskList(context);
		logger.info("NewsSpecialPalnJob div size{};regions:{};");
	}



	private void processTaskList(ShardingContext context) {
		Integer lastId = 0;
		do0: do {
			List<SpeclNewsPlan> queryPaging = newsSpeclPlanDao.queryPaging(lastId);
			if (queryPaging == null || queryPaging.isEmpty()) {
				break do0;
			}
			lastId = queryPaging.get(queryPaging.size()-1).getId();

			/*
			 * 获取本次执行的list
			 */
			List<SpeclNewsPlan> shardingPlan = ElasticJobUtil.shardingPlan(context, queryPaging);
			if (shardingPlan != null && !shardingPlan.isEmpty()) {

				shardingPlan.stream().filter(v -> checkSplan(v)).forEach(v -> {
					startDbSplan(v);
					SplanTask ta = new SplanTask();
					BeanUtils.copyProperties(v, ta);
					taskExecutor.execute(new RunableTaskSpan(ta));
				});
			}
		} while (true);
	}

	/*
	 * check 这个plan在本次是否应该被执行
	 */
	private boolean checkSplan(SpeclNewsPlan plan) {
		
		if (!StringUtils.equalsIgnoreCase(plan.getBranch(), carwlerJobStateUtil.getCarwler()))
			return false;
		/*
		 * 时间条件
		 */
		boolean a1 = Optional.ofNullable(plan.getNextTime()).map(v -> v.before(new Date()))
				.orElse(plan.getNextTime() == null);
		/*
		 * 状态条件1
		 */
		boolean a2 = Optional.ofNullable(plan.getExecStat())
				.map(v -> SpeclNewsPlan.ExecStat.done.equals(v))
				.orElse(plan.getExecStat() == null);

		/*
		 * 状态条件2
		 */
		boolean a3 = Optional.ofNullable(plan.getNextTime())
				.map(v -> new Date().after(DateUtils.addMinutes(v, plan.getCarshPeriod()))).orElse(false);
				 

		return a1 && (a2 || a3);
	}


	/**
	 * 回写NewsSpeclPlan状态，供下次job扫描使用
	 * 
	 * @param task
	 */
	private void writeDbSplan(SplanTask task) {
		SpeclNewsPlan plan = new SpeclNewsPlan();
		plan.setId(task.getId());
		plan.setExecStat(task.getExecStat());
		plan.setExecCount(1);
		newsSpeclPlanDao.update(plan);
	}

	private void doingDbSplan(SplanTask task) {
		SpeclNewsPlan plan = new SpeclNewsPlan();
		plan.setId(task.getId());
		plan.setExecStat(SpeclNewsPlan.ExecStat.doing);
		plan.setExecTime(new Date());
		newsSpeclPlanDao.update(plan);
	}

	private void startDbSplan(SpeclNewsPlan task) {
		SpeclNewsPlan plan = new SpeclNewsPlan();
		plan.setId(task.getId());
		plan.setExecStat(SpeclNewsPlan.ExecStat.start);

		CronSequenceGenerator cronenerator = new CronSequenceGenerator(task.getCron());
		Date nextTime = cronenerator.next(new Date());

		plan.setNextTime(nextTime);
		newsSpeclPlanDao.update(plan);
	}


	private class RunableTaskSpan implements Runnable {

		private SplanTask splanTask;

		public RunableTaskSpan(SplanTask splanTask) {
			this.splanTask = splanTask;
		}

		@Override
		public void run() {
			try {
				doingDbSplan(splanTask);
				splanHandlerMapping.process(splanTask);
			} finally {
				writeDbSplan(splanTask);
			}

		}

	}



}
